package ws

import (
	"encoding/json"
	"errors"
	"github.com/gorilla/websocket"
	"go.uber.org/zap"
	"io"
	"sync"
	"time"
)

const (
	// Time allowed to write a message to the peer.
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer.
	pongWait = 3 * time.Second

	// Send pings to peer with this period. Must be less than pongWait.
	pingPeriod = (pongWait * 9) / 10

	// Maximum message size allowed from peer.
	maxMessageSize = 1024
)

var ErrClosed = errors.New("connection 已关闭")

type Action string

type Msg struct {
	Action  Action `json:"action"`
	Payload any    `json:"payload,omitempty"`
}

type MsgHandle func(clientId string, msg *Msg)

type client struct {
	manage    *Manage
	log       *zap.Logger
	conn      *websocket.Conn
	msgChan   chan *Msg
	closeChan chan byte
	closeOnce sync.Once
	group     Group
	id        string
}

func (c *client) ID() string {
	if c.id == "" {
		c.id = c.conn.RemoteAddr().String() + "->" + c.conn.LocalAddr().String()
	}
	return c.id
}

func (c *client) send(msg *Msg) error {
	select {
	case c.msgChan <- msg:
	case <-c.closeChan:
		return ErrClosed
	}
	return nil
}

func (c *client) procLoop() {
	go c.readLoop()
	go c.writeLoop()
}

func (c *client) Close() (err error) {
	c.manage.Unregister(c.ID())
	c.closeOnce.Do(func() {
		close(c.closeChan)
		err = c.conn.Close()
	})
	return
}

func (c *client) readLoop() {
	defer c.Close()
	c.conn.SetReadLimit(maxMessageSize)
	for {
		mt, msg, err := c.conn.ReadMessage()
		c.log.Debug("wsConn readMessage", zap.Int("messageType", mt), zap.ByteString("msg", msg), zap.Error(err))
		if err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
				c.log.Error("errorIsUnexpectedCloseError: " + err.Error())
				return
			}
			c.log.Error("ReadMessage error", zap.Error(err))
			return
		}
		if mt != websocket.TextMessage {
			continue
		}
		_wsMsg := new(Msg)
		if err = json.Unmarshal(msg, _wsMsg); err != nil {
			c.log.Error("dealMsg json.Unmarshal(msg) error", zap.ByteString("msg", msg), zap.Error(err))
			continue
		}
		go c.manage.msgHandle(c.ID(), _wsMsg)
	}
}

func (c *client) writeLoop() {
	ticker := time.NewTicker(pingPeriod)
	var err error
	var w io.WriteCloser
	defer func() {
		if err != nil {
			c.log.Error("writeLoop error", zap.Error(err))
		}
		ticker.Stop()
		c.Close()
	}()
	send := func(w io.WriteCloser, msg *Msg) (n int, err error) {
		_msg, err := json.Marshal(msg)
		if err != nil {
			return
		}
		return w.Write(_msg)
	}
	for {
		select {
		case msg, ok := <-c.msgChan:
			_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if !ok {
				// The hub closed the channel.
				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
				return
			}
			w, err = c.conn.NextWriter(websocket.TextMessage)
			if err != nil {
				return
			}
			if _, err = send(w, msg); err != nil {
				return
			}
			// Add queued chat messages to the current websocket message.
			n := len(c.msgChan)
			for i := 0; i < n; i++ {
				if _, err = send(w, <-c.msgChan); err != nil {
					return
				}
			}
			if err = w.Close(); err != nil {
				return
			}
		case <-ticker.C:
			_ = c.conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return
			}
		}
	}
}
